import java.io.IOException;
import java.util.Arrays;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.pdfbox.pdmodel.PDDocument;
import org.apache.pdfbox.util.PDFTextStripper;

public class PDFLineRecordReader extends RecordReader<Text, Text> {

private String[] lines = null;
private LongWritable lineCounter = null;
private Text key = null;
private Text value = null;
private String fileYear;

@Override
public void initialize(InputSplit genericSplit, TaskAttemptContext context)
		throws IOException, InterruptedException {

	FileSplit split = (FileSplit) genericSplit;
	Configuration job = context.getConfiguration();
	final Path file = split.getPath();
	

	fileYear =file.getName().substring(9, 13);
	/*
	 * The below code contains the logic for opening the file and seek to
	 * the start of the split. Here we are applying the Pdf Parsing logic
	 */

	FileSystem fs = file.getFileSystem(job);
	FSDataInputStream fileIn = fs.open(split.getPath());
	PDDocument pdf = null;
	String parsedText = null;
	PDFTextStripper stripper;
	pdf = PDDocument.load(fileIn);
	stripper = new PDFTextStripper();
	parsedText = stripper.getText(pdf);
	this.lines = parsedText.split("\n");	
	}

@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
	
	if (lineCounter == null) {
		lineCounter = new LongWritable();
		lineCounter.set(1);
		value = new Text();
		value.set(lines[0]);
		key = new Text();
		key.set("2000");
	} else {
		int temp = (int) lineCounter.get();
		if (temp < (lines.length - 1)) {
			int count = (int) lineCounter.get();
			value = new Text();
			value.set(lines[count]);
			count = count + 1;
			lineCounter = new LongWritable(count);
			key = new Text(fileYear);
		} else {
			return false;
		}

	}
	if (lineCounter == null || value == null) {
		return false;
	} else {
		return true;
	}
	
}

/*@Override
public LongWritable getCurrentKey() throws IOException,
		InterruptedException {

	return key;
}
*/
@Override
public Text getCurrentValue() throws IOException, InterruptedException {

	return value;
}

@Override
public float getProgress() throws IOException, InterruptedException {

	return 0;
}

@Override
public void close() throws IOException {

}

@Override
public Text getCurrentKey() throws IOException, InterruptedException {
	// TODO Auto-generated method stub
	return key;
}

}
